package Creating_Observables

import (
	"context"
	"github.com/reactivex/rxgo/v2"
)

func Demo_OperatorOptions() {
	observable := rxgo.Create([]rxgo.Producer{func(ctx context.Context, next chan<- rxgo.Item) {
		next <- rxgo.Of(1)
		next <- rxgo.Of(2)
		next <- rxgo.Of(3)
	}})

	// 设置1 输出管道缓冲容量 1
	op := rxgo.WithBufferedChannel(1)
	//设置2 带有上下文
	//op2:=rxgo.WithContext(ctx)
	// 设置3 观察策略 默认rxgo.Lazy 当观察者被订阅的时候消费 rxgo.Eager 当观察者被创建的时候消费
	//_:=rxgo.WithObservationStrategy(rxgo.Lazy) //default
	// 设置4 错误策略 默认rxgo.StopOnError 当被观察者产出异常停止运行 rxgo.ContinueOnError 当被观察者产出异常继续运行
	//_:=rxgo.WithErrorStrategy(rxgo.StopOnError) //default
	// 设置5 设置并行协程数量
	//_:=rxgo.WithPool(8) //创建一个容量为8的协程池
	//_:=rxgo.WithCPUPool() //创建一个容量为runtime.NumCPU()的协程池
	// 在使用协程池下可以设置
	//_:=rxgo.Serialize(对象排序方法) // 有序的
	// 设置6 发布策略
	//_:=rxgo.WithPublishStrategy() //订阅者在被被观察者执行connect之后得到数据的副本并执行

	observable.Map(func(_ context.Context, i interface{}) (interface{}, error) {
		return i.(int) * 10, nil
	}, op)
}
